package cn.doitedu.eagle.functions;

import cn.doitedu.eagle.beans.CanalBinlog;
import cn.doitedu.eagle.beans.CanalRuleRecord;
import cn.doitedu.eagle.beans.Keyed;
import cn.doitedu.eagle.beans.LogBean;
import cn.doitedu.eagle.ruleutils.KeyExtractor;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.*;

/***
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contack qq:657270652 wx:doit_edu
 * @site www.51doit.cn
 * @date 2021/3/25
 * @desc 动态keyby分组核心实现函数组件
**/
public class DynamicKeyFunction extends BroadcastProcessFunction<LogBean, String, Keyed<LogBean,String>> {

    //分组Key,{rule-01,rule-02,rule-03}
    MapState<String, Set<String>> ruleKeyState;

    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor<String, Set<String>> ruleKyesStateDesc = new MapStateDescriptor<String,Set<String>>("groupingKeys", TypeInformation.of(new TypeHint<String>(){}),TypeInformation.of(new TypeHint<Set<String>>(){}));
        ruleKeyState = getRuntimeContext().getMapState(ruleKyesStateDesc);
    }

    @Override
    public void processElement(LogBean bean, ReadOnlyContext ctx, Collector<Keyed<LogBean, String>> out) throws Exception {

        Iterable<String> keySets = ruleKeyState.keys();
        for(String keys : keySets) {
            String key = KeyExtractor.extractKes(bean,keys);
            Keyed<LogBean, String> keyed = new Keyed<>(bean, key);
            out.collect(keyed);
        }
    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<Keyed<LogBean, String>> out) throws Exception {

        CanalBinlog canalBinlog = JSON.parseObject(value, CanalBinlog.class);
        CanalRuleRecord canalRuleRecord = canalBinlog.getData().get(0);


        String groupingKeys = canalRuleRecord.getGroupingKeys();
        String[] split = groupingKeys.split(",");
        Arrays.sort(split);
        String groupingKey = StringUtils.join(split, ",");

        Set<String> ruleIdSet = ruleKeyState.get(groupingKey);
        if(ruleIdSet == null || ruleIdSet.size()<1){
            ruleIdSet = new HashSet<>();
        }
        ruleIdSet.add(canalRuleRecord.getRule_name());

        ruleKeyState.put(groupingKey,ruleIdSet);
    }
}
